package zload

import (
	"context"
	"encoding/json"
	"sync"
	"time"

	"gitee.com/youkelike/ziam/zerrors"
	"gitee.com/youkelike/ziam/zredis"
	"gitee.com/youkelike/zlog"
	"github.com/go-redis/redis/v7"
)

// 用于处理 redis 的订阅，收到订阅信号后，调用 Reload 重载缓存数据

type Loader interface {
	Reload() error
}

type Load struct {
	// 外部传入的带 cancel 的 context，用于在外部控制 reloadLoop
	ctx context.Context
	// 重载数据时加锁
	lock   sync.Mutex
	loader Loader

	// 消息放入队列时加锁
	reloadQueueLock sync.Mutex
	reloadQueue     []func()
}

func NewLoad(ctx context.Context, loader Loader) *Load {
	return &Load{
		ctx:         ctx,
		loader:      loader,
		reloadQueue: make([]func(), 0, 4),
	}
}

func (l *Load) Start() {
	// 订阅频道
	// 这里没有在收到订阅消息后立即执行重载操作，而是把重载信号放入处理队列中，然后每隔一定时间去检查处理队列，
	// 如果处理队列不为空才执行重载操作，
	// 这样可以防止订阅消息到达太频繁时造成的重载并发压力
	go l.startPubSubLoop()
	// 启动消费者
	// 定时从处理队列中读取消息，调用数据重载操作
	go l.reloadLoop()
	// 手动调用，加载初始数据
	l.DoReload()
}

// 启动订阅功能
func (l *Load) startPubSubLoop() {
	redisStore := zredis.RedisCluster{}
	// redisStore.Connect()

	for {
		// 收到的订阅消息会传给回调函数
		err := redisStore.StartPubSubHandler(context.Background(), RedisPubSubChannel, func(v interface{}) {
			if l.ctx.Err() != nil {
				return
			}
			l.handleRedisEvent(v)
		})
		if err != nil {
			if !zerrors.Is(err, zredis.ErrRedisIsDown) {
				zlog.Errorw("connect to redis failed, reconnect in 10s: %s", err.Error())
			}

			time.Sleep(10 * time.Second)
			zlog.Warnw("reconnection: %s", err.Error())
		}
	}
}

// 解读收到的订阅消息，先放到队列
func (l *Load) handleRedisEvent(v interface{}) {
	message, ok := v.(*redis.Message)
	if !ok {
		return
	}

	notif := Notification{}
	if err := json.Unmarshal([]byte(message.Payload), &notif); err != nil {
		zlog.Errorw("unmarshal message body failed, malformed: ", err)
		return
	}

	zlog.Infow("receive redis message", "command", notif.Command, "payload", message.Payload)

	switch notif.Command {
	case NoticePolicyChanged, NoticeSecretChanged:
		zlog.Infow("reloading secrets and policies")
		// 这里实际往队列里扔的都是 nil
		l.reloadQueueLock.Lock()
		l.reloadQueue = append(l.reloadQueue, nil)
		l.reloadQueueLock.Unlock()
	default:
		zlog.Warnw("unknown notification command: %q", notif.Command)
		return
	}
}

// 一次性获取处理队列中的所有消息
func (l *Load) shouldReload() ([]func(), bool) {
	l.reloadQueueLock.Lock()
	defer l.reloadQueueLock.Unlock()

	if len(l.reloadQueue) == 0 {
		return nil, false
	}

	n := l.reloadQueue
	l.reloadQueue = []func(){}
	return n, true
}

// 数据流向 queue --> 处理，
// 定时任务，一次性把 queue 读空，也只需执行一次 Reload
func (l *Load) reloadLoop() {
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-l.ctx.Done():
			// 被外部叫停了
			return
		case <-ticker.C:
			_, ok := l.shouldReload()
			if !ok {
				continue
			}

			start := time.Now()
			l.DoReload()

			zlog.Infow("reload completed in %v", time.Since(start))
		}
	}
}

// 重新加载数据
func (l *Load) DoReload() {
	l.lock.Lock()
	defer l.lock.Unlock()

	if err := l.loader.Reload(); err != nil {
		zlog.Errorw("failed to reload storage: %s", err.Error())
	}

	zlog.Debugw("reload storage success")
}
